package com.example.demo.message;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.example.demo.common.Function;
import com.example.demo.config.MqttClientON;
import com.example.demo.config.MqttSender;
import com.example.demo.consts.DeviceTopicConstants;
import com.example.demo.domain.*;
import com.example.demo.websocket.SumoWebsocket;
import com.hiacent.vrs.gateway.tcsae157.MessageFrame;
import com.hiacent.vrs.gateway.tcsae157.Scene;
import com.hiacent.vrs.gateway.tcsae157.frame.BasicSafetyMessage;
import com.hiacent.vrs.gateway.tcsae157.frame.RoadSideInformation;
import com.hiacent.vrs.gateway.tcsae157.frame.RoadsideSafetyMessage;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
@EnableScheduling
public class FiveMessageConsumer {

    public static final String MESSAGE_KEY_SPAT = "spat:quene"; // 信号灯
    public static final String MESSAGE_KEY_RSM = "rsm:quene"; // bsm
    public static final String MESSAGE_KEY_BSM = "bsm:quene"; // bsm
    public static final String MESSAGE_KEY_RSI_WATER = "water:quene"; // 积水
    public static final String MESSAGE_KEY_RSI_LIMIT = "limit:quene"; // 限速
    public static final String MESSAGE_KEY_RSI_BUILD = "build:quene"; // 障碍物
    public static final String MESSAGE_KEY_RSI_LINE = "line:quene"; // 压线
//    public static final String MESSAGE_KEY_RSI_TRAFFIC_LIGHT = "traffic_light:quene"; // 闯红灯
    public static final String MESSAGE_KEY_RSI_TRAFFIC_ACCIDENT = "traffic_accident:quene"; // 前向碰撞

    public static final String MESSAGE_KEY_WARNING = "warn:quene"; // 闯红灯和限速提醒预警

    /**
     * 注入发送MQTT的Bean
     */
    @Resource
    private MqttSender MqttSender;

    @Autowired
    private RedisTemplate<String,Object> redisTemplate;

    public static final MqttClient client = MqttClientON.createClient();

    @Autowired
    private SumoWebsocket sumoWebsocket;
    AtomicInteger bsmCount = new AtomicInteger(0);
    AtomicInteger rsmCount = new AtomicInteger(0);
    AtomicInteger rsiCount = new AtomicInteger(0);
    AtomicInteger spatCount = new AtomicInteger(0);

    AtomicInteger warnCount = new AtomicInteger(0);
    //@Scheduled(initialDelay = 5 * 1000, fixedRate = 2 * 1000)
//    public void rPop() {
//        String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY);
//        log.info(message);
//    }

    @PostConstruct
    public void brPop() {
        new Thread(() -> {
            while (true) {
                try {
                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_RSM, 10, TimeUnit.SECONDS);
                    if (message == null) {
                        log.info("暂无RSM事件");
                    }
                    RSM rsm = JSONObject.parseObject(message, RSM.class);
                    if (rsm != null) {
//                        System.out.println("===============RSM===============");
                        Function function = new Function();
                        RoadsideSafetyMessage rsmEvent = function.getRsmEvent(message, rsm);
                        MessageFrame messageFrame = new MessageFrame();
                        messageFrame.setRsmFrame(rsmEvent);
                        rsmEvent.setMsgCount(rsmCount.addAndGet(1));
                        String msg = JSONObject.toJSONString(messageFrame, SerializerFeature.WriteMapNullValue);
                        String topic = String.format(DeviceTopicConstants.RSCU_V2X_UP, rsmEvent.getId());
                        MqttSender.sendToMqtt(topic, msg);
                        sumoWebsocket.pushMessage(msg);
//                        System.out.println(messageFrame);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }).start();

        new Thread(() -> {
            while (true) {
                try {
                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_BSM, 10, TimeUnit.SECONDS);
                    if (message == null) {
                        log.info("暂无BSM事件");
                    }
                    BSM bsm = JSONObject.parseObject(message, BSM.class);
                    if (bsm != null) {
//                        System.out.println("===============BSM===============");
                        Function function = new Function();
                        BasicSafetyMessage bsmEvent = function.getBsmEvent(message, bsm);
                        MessageFrame messageFrame = new MessageFrame();
                        messageFrame.setBsmFrame(bsmEvent);
                        bsmEvent.setMsgCnt(bsmCount.addAndGet(1));
                        String msg = JSONObject.toJSONString(messageFrame, SerializerFeature.WriteMapNullValue);
                        String topic = String.format(DeviceTopicConstants.RSCU_V2X_UP, bsmEvent.getId());
                        MqttSender.sendToMqtt(topic, msg);
                        sumoWebsocket.pushMessage(msg);
//                        System.out.println(messageFrame);
                    }
                } catch (Exception e) {
                    log.info(e.getMessage(), e);
                }
            }
        }).start();

//        new Thread(() -> {
//            while (true) {
//                try {
//                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_RSI_WATER, 10, TimeUnit.SECONDS);
//                    if (message == null) {
//                        log.info("暂无湿滑事件");
//                    }
//                    RtesRSI rtesRSI = JSONObject.parseObject(message, RtesRSI.class);
//                    if (rtesRSI != null) {
//                        System.out.println("===============water===============");
//                        Function function = new Function();
//                        RoadSideInformation rtesEvent = function.getRtesEvent(message, rtesRSI);
//                        MessageFrame messageFrame = new MessageFrame();
//                        messageFrame.setRsiFrame(rtesEvent);
//                        rtesEvent.setMsgCount(rsiCount.addAndGet(1));
//                        String msg = JSONObject.toJSONString(messageFrame, SerializerFeature.WriteMapNullValue);
//                        String topic = String.format(DeviceTopicConstants.RSCU_V2X_UP, rtesEvent.getId());
//                        MqttSender.sendToMqtt(topic, msg);
//                        sumoWebsocket.pushMessage(msg);
//                        System.out.println(messageFrame);
//                    }
//                } catch (Exception e) {
//                    log.info(e.getMessage(), e);
//                }
//            }
//        }).start();

        new Thread(() -> {
            while (true) {
                try {
                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_RSI_LIMIT, 10, TimeUnit.SECONDS);
                    if (message == null) {
                        log.info("暂无超速事件");
                    }
                    RtssRSI rtssRSI = JSONObject.parseObject(message, RtssRSI.class);
                    if (rtssRSI != null) {
                        System.out.println("===============xiansu===============");
                        Function function = new Function();
                        RoadSideInformation rtssEvent = function.getRtssEvent(message, rtssRSI);
                        MessageFrame messageFrame = new MessageFrame();
                        messageFrame.setRsiFrame(rtssEvent);
                        rtssEvent.setMsgCount(rsiCount.addAndGet(1));
                        String msg = JSONObject.toJSONString(messageFrame, SerializerFeature.WriteMapNullValue);
                        String topic = String.format(DeviceTopicConstants.RSCU_V2X_UP, rtssEvent.getId());
                        MqttClientON.clientON(msg, topic, client);
                        sumoWebsocket.pushMessage(msg);
                        System.out.println(msg);
                    }
                } catch (Exception e) {
                    log.info(e.getMessage(), e);
                }
            }
        }).start();

        new Thread(() -> {
            while (true) {
                try {
                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_RSI_BUILD, 10, TimeUnit.SECONDS);
                    if (message == null) {
                        log.info("暂无障碍物事件");
                    }
                    RtesRSI rtesRSI = JSONObject.parseObject(message, RtesRSI.class);
                    if (rtesRSI != null) {
                        System.out.println("===============barrier===============");
                        Function function = new Function();
                        RoadSideInformation rtesEvent = function.getRtesEvent(message, rtesRSI);
                        MessageFrame messageFrame = new MessageFrame();
                        messageFrame.setRsiFrame(rtesEvent);
                        rtesEvent.setMsgCount(rsiCount.addAndGet(1));
                        String msg = JSONObject.toJSONString(messageFrame, SerializerFeature.WriteMapNullValue);
                        String topic = String.format(DeviceTopicConstants.RSCU_V2X_UP, rtesEvent.getId());
                        MqttClientON.clientON(msg, topic, client);
//                        MqttSender.sendToMqtt(topic, msg);
                        sumoWebsocket.pushMessage(msg);
                        System.out.println(messageFrame);
                    }
                } catch (Exception e) {
                    log.info(e.getMessage(), e);
                }
            }
        }).start();

//        new Thread(() -> {
//            while (true) {
//                try {
//                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_RSI_LINE, 10, TimeUnit.SECONDS);
//                    if (message == null) {
//                        log.info("暂无压线事件");
//                    }
//                    RtesRSI rtesRSI = JSONObject.parseObject(message, RtesRSI.class);
//                    if (rtesRSI != null) {
//                        System.out.println("===============yaxian===============");
//                        Function function = new Function();
//                        RoadSideInformation rtesEvent = function.getRtesEvent(message, rtesRSI);
//                        MessageFrame messageFrame = new MessageFrame();
//                        messageFrame.setRsiFrame(rtesEvent);
//                        rtesEvent.setMsgCount(rsiCount.addAndGet(1));
//                        String msg = JSONObject.toJSONString(messageFrame, SerializerFeature.WriteMapNullValue);
//                        String topic = String.format(DeviceTopicConstants.RSCU_V2X_UP, rtesEvent.getId());
//                        MqttSender.sendToMqtt(topic, msg);
//                        sumoWebsocket.pushMessage(msg);
//                        System.out.println(messageFrame);
//                    }
//                } catch (Exception e) {
//                    log.info(e.getMessage(), e);
//                }
//            }
//        }).start();

//        new Thread(() -> {
//            while (true) {
//                try {
//                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_RSI_TRAFFIC_LIGHT, 10, TimeUnit.SECONDS);
//                    if (message == null) {
//                        log.info("暂无闯红灯事件");
//                    }
//                    RtssRSI rtssRSI = JSONObject.parseObject(message, RtssRSI.class);if (rtssRSI != null) {
//                        System.out.println("===============trafficLight===============");
//                        Function function = new Function();
//                        RoadSideInformation rtssEvent = function.getRtssEvent(message, rtssRSI);
//                        MessageFrame messageFrame = new MessageFrame();
//                        messageFrame.setRsiFrame(rtssEvent);
//                        rtssEvent.setMsgCount(rsiCount.addAndGet(1));
//                        String msg = JSONObject.toJSONString(messageFrame, SerializerFeature.WriteMapNullValue);
//                        String topic = String.format(DeviceTopicConstants.RSCU_V2X_UP, rtssEvent.getId());
//                        MqttSender.sendToMqtt(topic, msg);
//                        sumoWebsocket.pushMessage(msg);
//                        System.out.println(messageFrame);
//                    }
//                } catch (Exception e) {
//                    log.info(e.getMessage(), e);
//                }
//            }
//        }).start();

        new Thread(() -> {
            while (true) {
                try {
                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_RSI_TRAFFIC_ACCIDENT, 10, TimeUnit.SECONDS);
                    if (message == null) {
                        log.info("暂无交通事故");
                    }
                    RtesRSI rtesRSI = JSONObject.parseObject(message, RtesRSI.class);
                    if (rtesRSI != null) {
                        System.out.println("===============trafficAccident===============");
                        Function function = new Function();
                        RoadSideInformation rtesEvent = function.getRtesEvent(message, rtesRSI);
                        MessageFrame messageFrame = new MessageFrame();
                        messageFrame.setRsiFrame(rtesEvent);
                        rtesEvent.setMsgCount(rsiCount.addAndGet(1));
                        String msg = JSONObject.toJSONString(messageFrame, SerializerFeature.WriteMapNullValue);
                        String topic = String.format(DeviceTopicConstants.RSCU_V2X_UP, rtesEvent.getId());
                        MqttClientON.clientON(msg, topic, client);
//                        MqttSender.sendToMqtt(topic, msg);
                        sumoWebsocket.pushMessage(msg);
                        System.out.println(messageFrame);
                    }
                } catch (Exception e) {
                    log.info(e.getMessage(), e);
                }
            }
        }).start();

        new Thread(() -> {
            while (true) {
                try {
                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_SPAT, 10, TimeUnit.SECONDS);
                    if (message == null) {
                        log.info("暂无红绿灯上报");
                    }
                    SPAT spatMsg = JSONObject.parseObject(message, SPAT.class);
                    if (spatMsg != null) {
                        System.out.println("===============spat===============");
                        Function function = new Function();
                        com.hiacent.vrs.gateway.tcsae157.frame.SPAT spat = function.getSpatMsg(message, spatMsg);
                        MessageFrame messageFrame = new MessageFrame();
                        messageFrame.setSpatFrame(spat);
                        spat.setMsgCount(spatCount.addAndGet(1));
                        String msg = JSONObject.toJSONString(messageFrame, SerializerFeature.WriteMapNullValue);
                        String topic = String.format(DeviceTopicConstants.RSCU_V2X_UP, spat.getId());
                        MqttSender.sendToMqtt(topic, msg);
                        sumoWebsocket.pushMessage(msg);
                        System.out.println(messageFrame);
                    }
                } catch (Exception e) {
                    log.info(e.getMessage(), e);
                }
            }
        }).start();

        new Thread(() -> {
            while (true) {
                try {
                    String message = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY_WARNING, 10, TimeUnit.SECONDS);
                    if (message == null) {
                        log.info("暂无预警上报");
                    }
                    Sceneing scene = JSONObject.parseObject(message, Sceneing.class);
                    if (scene != null) {
                        System.out.println("===============warn===============");
                        Function function = new Function();
                        Scene sc = function.getWarnMsg(message, scene);
                        String warnMsg = JSONObject.toJSONString(sc, SerializerFeature.WriteMapNullValue);
                        String topic = String.format(DeviceTopicConstants.WARNING_EVENT_UP, scene.getLicense_plate_number());
                        MqttSender.sendToMqtt(topic, warnMsg);
                        warnCount.addAndGet(1);
                        System.out.println(warnMsg);
                    }
                } catch (Exception e) {
                    log.info(e.getMessage(), e);
                }
            }
        }).start();

        new Thread(() -> {
            while (true) {
                try {
                    int i = bsmCount.get();
                    int i1 = rsmCount.get();
                    int i2 = rsiCount.get();
                    int i3 = spatCount.get();
                    int i4 = warnCount.get();
                    int all = i + i1 + i2 + i3;
                    if (all == 0) {
                        log.info("四大消息总数为0");
                    }else {
                        System.out.println("bsm消息数为"+ i);
                        System.out.println("rsm消息数为"+ i1);
                        System.out.println("rsi消息数为"+ i2);
                        System.out.println("spat消息数为"+ i3);
                        System.out.println("warn消息数为"+ i4);
                        System.out.println("四大消息总数为" + (i + i1 + i2 + i3));
                    }
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();
    }
}